package com.ww.hadoop.mr.topn_mapjoin;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;

@Slf4j
public class TMapper extends Mapper<LongWritable, Text, TKey, IntWritable> {
    static final Logger logger = LoggerFactory.getLogger(TMapper.class);

    DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    TKey mkey = new TKey();
    IntWritable mval = new IntWritable();

    HashMap<String,String>  dict = new HashMap<String,String>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {

        URI[] files = context.getCacheFiles();
        Path path = new Path(files[0].getPath());

        BufferedReader reader = new BufferedReader(new FileReader(new File(path.getName())));

        String line = reader.readLine();

        while(line != null){
            String[] split = line.split("\t");
            dict.put(split[0],split[1]);
            line = reader.readLine();
        }


    }





    @SneakyThrows
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, TKey, IntWritable>.Context context) throws IOException, InterruptedException {
        //2019-6-1 22:22:22	1	31
        log.info("value 的值为:{}", value);

        String[] split = StringUtils.split(value.toString(),'\t');
        log.info("split 的值为:{}", split);
        LocalDateTime d = LocalDateTime.parse(split[0], dtf);
        mkey.setYear(d.getYear());
        mkey.setMonth(d.getMonthValue());
        mkey.setDay(d.getDayOfMonth());
        int wd = Integer.parseInt(split[2]);
        mkey.setWd(wd);
        mkey.setAddress(dict.get(split[1]));
        mval.set(wd);
        context.write(mkey, mval);
    }

}
